package com.flc.archive.writer;

import com.flc.archive.hdfs.CopyLocalFileToHdfs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @author haizi
 * @date 2018/2/2
 */
public class DataWriter {

    private static final Logger logger = LoggerFactory.getLogger(DataWriter.class);

    private int count = 0;
    private int flushNum = 1000;
    private String filePath = "";
    private String fileName = "";
    private String hdfsPath = "";
    private BufferedWriter bufferedWriter = null;

    public DataWriter(int flushNum, String filePath, String hdfsPath) throws IOException {
        if (flushNum > 0) {
            this.flushNum = flushNum;
        }
        if (!filePath.isEmpty() && !hdfsPath.isEmpty()) {
            this.filePath = filePath;
            this.hdfsPath = hdfsPath;

            this.bufferedWriter = this.getBW();
        }
        logger.info("打开路径" + filePath + ", 准备在该路径下按日期分类写入");
    }

    private BufferedWriter getBW() throws IOException {
        File dir = new File(filePath);
        if(!dir.exists()) {
            dir.mkdirs();
        }
        Date today = new Date();
        String dateAsFileName = new SimpleDateFormat("yyyy-MM-dd").format(today);

        fileName =  dateAsFileName + ".txt";

        File f = new File(filePath + fileName);
        if (!f.exists()) {
            f.createNewFile();
        }
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(f, true)));
        return bw;
    }

    /**
     * 写入数据
     * @param dataString 待写入的轨迹数据
     * @throws IOException 文件IO的异常，理论上不会出现了。
     */
    public void write(String dataString) throws IOException {
        bufferedWriter.write(dataString);
        // 定量刷盘，提高效率
        if (++count >= flushNum) {
            bufferedWriter.flush();
            count = 0;
        }

    }

    /**
     * 新的一天，换文件写入
     * @throws IOException 文件IO的异常，理论上不会出现了。
     */
    public void newDailyFile() throws IOException {

        // 刷盘，然后暂存待上传文件名和路径，原文件需要上传至HDFS
        String yesterdayFileName = filePath + fileName;
        bufferedWriter.flush();
        // 切换新的文件
        BufferedWriter lastWriter = this.bufferedWriter;
        this.bufferedWriter = getBW();
        // 刷盘（由切换文件可能产生部分数据未能正确写入）
        this.bufferedWriter.flush();
        lastWriter.close();
        // 开启线程来上传文件至HDFS
        CopyLocalFileToHdfs fileToHDFS = new CopyLocalFileToHdfs(yesterdayFileName, hdfsPath + fileName);
        Thread thread = new Thread(fileToHDFS);
        thread.start();
    }
}
